package pub.cleangao.memm.transport.netty;


import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.ReferenceCountUtil;
import lombok.extern.slf4j.Slf4j;
import pub.cleangao.memm.exception.RpcException;
import pub.cleangao.memm.registry.RpcRuntime;
import pub.cleangao.memm.transport.RpcRequestHandler;
import pub.cleangao.memm.transport.TransportConst;
import pub.cleangao.memm.transport.message.*;
import pub.cleangao.memm.transport.netty.compress.CompressType;
import pub.cleangao.memm.transport.netty.serialize.SerializationType;

import java.net.SocketException;


@Slf4j
public class NettyRpcServerHandler extends ChannelInboundHandlerAdapter {

    private final RpcRequestHandler rpcRequestHandler;

    public NettyRpcServerHandler() {
        this.rpcRequestHandler = new RpcRequestHandler();
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        RpcMessage rpcMessage = new RpcMessage();
        try {
            if (msg instanceof RpcMessage) {
                byte messageType = ((RpcMessage) msg).getMessageType();
                rpcMessage.setCodec(SerializationType.PROTOSTUFF.getCode());
                rpcMessage.setCompress(CompressType.NONE.getCode());
                if (messageType == MessageType.HEARTBEAT_REQUEST.value()) {
                    rpcMessage.setMessageType(messageType);
                    rpcMessage.setData(TransportConst.PONG);
                } else {
                    RpcRequest rpcRequest = (RpcRequest) ((RpcMessage) msg).getData();
                    try {
                        Object result = rpcRequestHandler.handle(rpcRequest);
                        rpcMessage.setMessageType(MessageType.RESPONSE.value());
                        if (ctx.channel().isActive() && ctx.channel().isWritable()) {
                            RpcResponse<Object> rpcResponse = RpcResponse.success(result, rpcRequest.getRequestId());
                            rpcMessage.setData(rpcResponse);
                        } else {
                            RpcResponse<Object> rpcResponse = RpcResponse.fail(rpcRequest.getRequestId());
                            rpcMessage.setData(rpcResponse);
                        }
                    } catch (RpcException e) {
                        rpcMessage.setMessageType(MessageType.EXCEPTION_RESPONSE.value());
                        RpcResponse<Object> rpcResponse = RpcResponse.fail(rpcRequest.getRequestId());
                        rpcResponse.setMessage(e.toString());
                        e.printStackTrace();
                        log.error("Method invoke runtime exception: {}", rpcResponse.getMessage());
                        // 是否存在降级处理
                        Object rst = RpcRuntime.invokeFallback(rpcRequest.getRpcServiceName(), rpcRequest.getMethodName(), rpcRequest.getParamTypes(), rpcRequest.getParameters());
                        rpcResponse.setData(rst);
                        rpcMessage.setData(rpcResponse);

                    }
                }
            }
            ctx.writeAndFlush(rpcMessage).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
        } finally {
            ReferenceCountUtil.release(msg);
        }
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleState state = ((IdleStateEvent) evt).state();
            if (state == IdleState.READER_IDLE) {
                ctx.close();
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        if (cause instanceof SocketException) {
            log.info("disconnected: " + ctx.name());
        } else {
            cause.printStackTrace();
        }
        ctx.close();

    }
}
